package cc.shacocloud.redis.study.queue.message;

import cc.shacocloud.redis.study.util.codeBlock.CodeBlockMD;
import cc.shacocloud.redis.study.util.codeBlock.CodeBlockMDManage;
import cc.shacocloud.redis.study.util.redis.RedisUtil;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.IdUtil;
import redis.clients.jedis.Builder;
import redis.clients.jedis.BuilderFactory;
import redis.clients.jedis.exceptions.JedisDataException;
import redis.clients.jedis.util.SafeEncoder;

import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 使用 Redis Stream 实现消息队列
 */
@CodeBlockMD(mdDirPath = "/redis/queue", mdFileName = "messageQueue_stream")
public class RedisStreamMessageQueue implements MessageQueue {

    private static final Class<RedisStreamMessageQueue> CLASS = RedisStreamMessageQueue.class;

    private final byte[] group;
    private final byte[] consumer;
    private final byte[] queueName;

    /**
     * 使用随机名字作为消费者名称
     */
    public RedisStreamMessageQueue(String queueName, String group) {
        this(queueName, group, IdUtil.objectId());
    }

    /**
     * @param queueName 消费队列名称
     * @param group     消费者组
     * @param consumer  消费者名称
     */
    public RedisStreamMessageQueue(String queueName, String group, String consumer) {
        this.group = SafeEncoder.encode(group);
        this.consumer = SafeEncoder.encode(consumer);
        this.queueName = SafeEncoder.encode(queueName);

        try {
            RedisUtil.tryFun(
                    jedis -> jedis.xgroupCreate(this.queueName, this.group, SafeEncoder.encode("0"), true),
                    "xgroup create"
            );
        } catch (JedisDataException e) {
            if ("BUSYGROUP Consumer Group name already exists".equals(e.getMessage())) {
                // 忽略
                return;
            }
            throw e;
        }
    }

    @Override
    public List<String> batchAddMessage(byte[]... dataArr) {
        String script = CodeBlockMDManage.get(CLASS, "batchAddMessage");

        // 拷贝封装为一个新的数组
        int len = dataArr.length;
        Object[] params = new Object[len + 2];
        params[0] = queueName;
        params[1] = String.valueOf(len);
        System.arraycopy(dataArr, 0, params, 2, len);

        return RedisUtil.eval(BuilderFactory.STRING_LIST, script, 1, params);
    }

    @SuppressWarnings("unchecked")
    public static final Builder<List<MessageConsumerMo.MessageMo>> STREAM_ENTRY_LIST =
            new Builder<List<MessageConsumerMo.MessageMo>>() {

                @Override
                public List<MessageConsumerMo.MessageMo> build(Object data) {
                    if (null == data) return null;

                    List<ArrayList<Object>> objectList = (List<ArrayList<Object>>) data;
                    if (objectList.isEmpty()) return null;

                    List<MessageConsumerMo.MessageMo> responses = new ArrayList<>(objectList.size() / 2);
                    for (ArrayList<Object> res : objectList) {

                        Iterator<byte[]> hashIterator = ((List<byte[]>) res.get(1)).iterator();
                        if (hashIterator.hasNext()) {
                            hashIterator.next();

                            MessageConsumerMo.MessageMo message = new MessageConsumerMo.MessageMo();
                            String id = SafeEncoder.encode((byte[]) res.get(0));
                            message.setMessageId(id);
                            message.setMessage(hashIterator.next());
                            responses.add(message);
                        }
                    }

                    return responses;
                }

                @Override
                public String toString() {
                    return "List<MessageConsumerMo.MessageMo>";
                }
            };

    @SuppressWarnings("unchecked")
    public static final Builder<List<MessageConsumerMo>> STREAM_DATA_LIST = new Builder<List<MessageConsumerMo>>() {

        @Override
        public List<MessageConsumerMo> build(Object data) {
            List<Object> streamsEntries = (List<Object>) data;
            if (CollUtil.isEmpty(streamsEntries)) return null;

            List<MessageConsumerMo> result = new ArrayList<>(streamsEntries.size());
            for (Object streamObj : streamsEntries) {
                List<Object> stream = (List<Object>) streamObj;
                if (stream == null) continue;

                List<MessageConsumerMo.MessageMo> messages = STREAM_ENTRY_LIST.build(stream.get(1));
                if (CollUtil.isEmpty(messages)) continue;

                MessageConsumerMo msgMo = new MessageConsumerMo();
                msgMo.setQueueName(SafeEncoder.encode((byte[]) stream.get(0)));
                msgMo.setMessages(messages);
                result.add(msgMo);
            }
            return result;
        }

        @Override
        public String toString() {
            return "List<MessageConsumerMo>";
        }
    };

    @Override
    public MessageConsumerMo batchGetMessage(int batchSize) {
        Map<byte[], byte[]> streams = new HashMap<>(2, 1);
        streams.put(queueName, SafeEncoder.encode("0"));

        List<MessageConsumerMo> result = RedisUtil.tryFun(
                jedis -> STREAM_DATA_LIST.build(jedis.xreadGroup(group, consumer, batchSize, 0L, false, streams)),
                "xread group");
        return CollUtil.isEmpty(result) ? null : result.get(0);
    }

    @Override
    public void ack(String id) {
        RedisUtil.tryFun(jedis -> jedis.xack(queueName, group, SafeEncoder.encode(id)));
    }

    private static final ExecutorService monitorService = Executors.newCachedThreadPool();

    private final List<MessageConsumer> consumers = new CopyOnWriteArrayList<>();

    @Override
    public synchronized void addConsumer(MessageConsumer consumer) {
        if (consumers.isEmpty()) {
            monitorService.execute(new MessageMonitor());
        }

        consumers.add(consumer);
    }

    /**
     * 消息监听
     */
    private class MessageMonitor implements Runnable {

        private final int count;
        private final long block;

        public MessageMonitor() {
            this(1, 60 * 1000L);
        }

        public MessageMonitor(int count, long block) {
            this.count = count;
            this.block = block;
        }

        @Override
        public void run() {
            Map<byte[], byte[]> streams = new HashMap<>(2, 1);
            streams.put(queueName, SafeEncoder.encode(">"));

            while (true) {
                // 阻塞式获取消息
                List<MessageConsumerMo> result = RedisUtil.tryFun(
                        jedis -> {
                            Object resultData = jedis.xreadGroup(group, consumer, count, block, false, streams);
                            return STREAM_DATA_LIST.build(resultData);
                        },
                        false, "xread group block", 0);

                // 分发
                if (CollUtil.isNotEmpty(result)) {
                    final MessageConsumerMo message = result.get(0);
                    for (MessageConsumer messageConsumer : consumers) {
                        messageConsumer.consumer(RedisStreamMessageQueue.this, message);
                    }
                }
            }
        }
    }

}
